package org.zzz.pcdc;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.sql.DataSource;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;


@Configuration
public class Config implements InitializingBean {
    public static final Logger LOGGER = LoggerFactory.getLogger(Config.class);
    @Autowired
    DataSource dataSource;


    @Override
    public void afterPropertiesSet() throws Exception {
        PcdcDataHandler dataHandler = (tc, r) -> {
            List<Object> idList = r.stream().map(m -> m.get(tc.getIdColumnName())).toList();
            try {
                //模拟实际的操作
                TimeUnit.MILLISECONDS.sleep(1);
                LOGGER.debug("已处理更新的数据: " + idList);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        };

        PcdcErrorHandler errorHandler = new PcdcErrorHandler() {
            @Override
            public void onError(PcdcTableConfig pcdcTableConfig, List<Map<String, Object>> data, Exception e) {
                List<Object> idList = data.stream()
                        .map(m -> m.get(pcdcTableConfig.getIdColumnName())).toList();
                LOGGER.error("已处理更新的数据: " + pcdcTableConfig.getCdcTableName() + " id= " + idList, e);
            }
        };

        String[] tables = new String[]{
               /* "zbench_cameras",
                "zbench_earphones",
                "zbench_laptops",
                "zbench_mobile_phones"*/
                "zbench_cameras"
        };

        for (String table : tables) {
            //模拟4个实例并发处理，每个实例5个工作线程
            for (int i = 0; i < 4; i++) {

                Pcdc pcdc = new PcdcBuilder(dataSource, table)
                        .updatedAtColumn("updated_at")
                        .idColumn("id")
                        .pollingIntervalSeconds(5)
                        .throttleSeconds(5)
                        .batchWorkSize(500)
                        .batchWorkTimeoutSeconds(300)
                        .partitionThreshold(10000)
                        .minPartitionIntervalSeconds(30)
                        .workerPoolSize(5)
                        .dataHandler(dataHandler)
                        .errorHandler(errorHandler)
                        .build();
                pcdc.startup();
            }
        }
    }
}
